package com.zhang.flink.example;

import com.zhang.flink.bean.OrderEvent;
import com.zhang.flink.bean.ReceiptEvent;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @title: 实时对账（不使用时间状态变量）
 * @author: zhang
 * @date: 2022/2/12 13:01
 */
public class OrderReceiptWithState {
    public static void main(String[] args) throws Exception {
        // todo 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //支付数据流
        SingleOutputStreamOperator<OrderEvent> orderEventDS = env
                .readTextFile("src/main/resources/OrderLog.csv")
                .map(new MapFunction<String, OrderEvent>() {
                    @Override
                    public OrderEvent map(String value) throws Exception {
                        String[] fields = value.split(",");
                        return new OrderEvent(
                                fields[0], fields[1], fields[2],
                                Long.parseLong(fields[3]) * 1000L
                        );
                    }
                })
                .filter(r -> "pay".equals(r.type))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderEvent>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {
                                    @Override
                                    public long extractTimestamp(OrderEvent element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                );

        //到账数据流
        SingleOutputStreamOperator<ReceiptEvent> receiptEventDS = env
                .readTextFile("src/main/resources/ReceiptLog.csv")
                .map(new MapFunction<String, ReceiptEvent>() {
                    @Override
                    public ReceiptEvent map(String value) throws Exception {
                        String[] fields = value.split(",");
                        return new ReceiptEvent(
                                fields[0], fields[1],
                                Long.parseLong(fields[2]) * 1000L
                        );
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<ReceiptEvent>forMonotonousTimestamps()
                                .withTimestampAssigner(
                                        new SerializableTimestampAssigner<ReceiptEvent>() {
                                            @Override
                                            public long extractTimestamp(ReceiptEvent element, long recordTimestamp) {
                                                return element.ts;
                                            }
                                        }
                                )
                );

        //对账逻辑
        SingleOutputStreamOperator<Tuple2<OrderEvent, ReceiptEvent>> result = orderEventDS
                .keyBy(r -> r.payId)
                .connect(receiptEventDS.keyBy(r -> r.payId))
                .process(new MatchOrder());

        //主流对账成功数据
        result.print("对账成功：");
        //提取侧输出流对账失败数据
        result.getSideOutput(new OutputTag<String>("no-receipt") {
        }).print("no-receipt");
        result.getSideOutput(new OutputTag<String>("no-order") {
        }).print("no-order");


        //todo 执行任务
        env.execute();
    }

    public static class MatchOrder extends CoProcessFunction<OrderEvent, ReceiptEvent, Tuple2<OrderEvent, ReceiptEvent>> {
        // 储存支付状态
        private ValueState<OrderEvent> orderEventState;
        // 储存到账数据
        private ValueState<ReceiptEvent> receiptEventState;

        @Override
        public void open(Configuration parameters) throws Exception {
            orderEventState = getRuntimeContext().getState(
                    new ValueStateDescriptor<OrderEvent>(
                            "order-state",
                            Types.POJO(OrderEvent.class)
                    )
            );

            receiptEventState = getRuntimeContext().getState(
                    new ValueStateDescriptor<ReceiptEvent>(
                            "receipt-state",
                            Types.POJO(ReceiptEvent.class)
                    )
            );
        }

        @Override
        public void processElement1(OrderEvent value, CoProcessFunction<OrderEvent, ReceiptEvent, Tuple2<OrderEvent, ReceiptEvent>>.Context ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
            if (receiptEventState.value() == null) {
                //支付数据到达，到账数据还未到，
                //1 储存支付状态
                orderEventState.update(value);
                //2 注册10s定时器等待到账数据到来
                ctx.timerService().registerEventTimeTimer(value.ts + 10 * 1000L);
            } else {
                //支付数据在到账数据等待的时间范围内来到，对账成功
                out.collect(Tuple2.of(value, receiptEventState.value()));
                //清除到账数据状态
                receiptEventState.clear();
            }
        }

        @Override
        public void processElement2(ReceiptEvent value, CoProcessFunction<OrderEvent, ReceiptEvent, Tuple2<OrderEvent, ReceiptEvent>>.Context ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
            if (orderEventState.value() == null) {
                //到账数据到达，支付数据还未到达
                // 1 储存到账状态
                receiptEventState.update(value);
                // 2 注册5s定时器等到支付数据到来
                ctx.timerService().registerEventTimeTimer(value.ts + 5 + 1000L);
            } else {
                //到账数据在支付数据等待的时间范围内到达，对账成功
                out.collect(Tuple2.of(orderEventState.value(), value));
                // 清除支付数据状态
                orderEventState.clear();
            }
        }

        @Override
        public void onTimer(long timestamp, CoProcessFunction<OrderEvent, ReceiptEvent, Tuple2<OrderEvent, ReceiptEvent>>.OnTimerContext ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
            if (orderEventState.value() != null) {
                //对账失败侧发送到输出流
                ctx.output(new OutputTag<String>("no-receipt") {
                }, orderEventState.value().payId + "只有支付数据没有到账数据");
                //清空状态
                orderEventState.clear();
            }

            if (receiptEventState.value() != null) {
                //对账失败侧发送到输出流
                ctx.output(new OutputTag<String>("no-order") {
                }, receiptEventState.value().payId + "只有到账数据没有支付数据");
                receiptEventState.clear();
            }


        }
    }
}
